#include "Peer.h"

Peer::Peer(string ip, int port, string deviceType){
    
    localInfo = new peerInfo();
    localInfo->IP = ip;
    localInfo->Port = port;
    localInfo->State = peerInfo::unknown;
    localInfo->deviceType = deviceType;

    // hardcoded
    _peerList = "peerList.txt";

    // initialize FileSystem...as empty
    VFS = new FileSystem();

    // set up mutexes
    sem_init (&job_queue_count, 0, 0);
    pthread_mutex_init (&job_queue_mutex, NULL);
    pthread_mutex_init (&VFS_mutex, NULL);
    pthread_mutex_init (&network_info_mutex, NULL);

    // set up threads
    startListenerThread();
    startWorkerThreads();

    // set up peer information - we don't know device type, or State
    processPeerList(_peerList);
}

// [**] delete any remaining job, destroy all threads, delete transit vector
Peer::~Peer(){
    // send an EXIT job to the listener and end it
    Job *job = new Job();
    job->type = Job::EXITLISTENER;
    job->args = NULL;
    pushJob(job);
    
    pthread_join(_listenerContext, NULL);
    DBG_PRINTF(("listener ended\n"));
    pthread_cancel(_listenerContext);

    // send EXIT job to all workers to end them
    for(int i = 0; i < maxWorkers; i++){
        Job *endworker = new Job();
        endworker->type = Job::EXITWORKER;
        endworker->args = NULL;
        pushJob(endworker); 
    }

    for (unsigned int ii = 0; ii < _workerContext.size(); ii++) {
        pthread_join(*_workerContext[ii], NULL);
    }

    // delete all thread pointers
    while (!_workerContext.empty()) {
        delete _workerContext.back();
        _workerContext.pop_back();
    }

    // destroy the semaphore
    sem_destroy (&job_queue_count);

    pthread_mutex_destroy(&job_queue_mutex);
    pthread_mutex_destroy(&VFS_mutex);
    pthread_mutex_destroy(&network_info_mutex);

    for (int ii = 0; ii < _numPeers; ii++) {
        delete _peers[ii];
    }

    delete VFS;
}


int Peer::list(){
    // call VFS to list the files
    VFS->list();

    return SAFE;
}

int Peer::join(){
    // acquire VFS mutex
    pthread_mutex_lock (&VFS_mutex);

    // determine all peer states + information about them (populate peerInfo with State and device type)
    // set own state to connected and broadcast -- combine join message with query message

    char msg[bufSize];
    int msgLen;

    // set status to be connected
    localInfo->State = peerInfo::connected;

    // broadcast join and initialize network information
    create_message(Peer::JOINANDQUERY, NULL, msg, msgLen);  
    blockingBroadcast(msg, msgLen);

    // determine one peer's VFS and update your own with it
    create_message(Peer::QUERYVFS, NULL, msg, msgLen); 
    blockingSendToAny(msg, msgLen);

    // release VFS mutex
    pthread_mutex_unlock (&VFS_mutex);

    if (localInfo->deviceType != "mobile") {
        startUpdaterThread();
    }

    printf("successfully joined the network\n");

    return SAFE;
}

int Peer::leave(){

    char msg[bufSize];
    int msgLen;

    if (localInfo->State != peerInfo::disconnected) {
        if (localInfo->deviceType != "mobile") {
            terminateUpdaterThread();
        }

        // leave message broadcast to everyone
        create_message(Peer::LEAVE, NULL, msg, msgLen);  
        nonblockingBroadcast(msg, msgLen);

        // transferHost: talk to 1 peer to give all files with latest revision
        create_message(Peer::QUERYTRANSFERFILES, NULL, msg, msgLen);  
        blockingSendToAny(msg, msgLen);     

        pthread_mutex_lock (&VFS_mutex);
        VFS->removeAllHosts();
        pthread_mutex_unlock (&VFS_mutex);

        // set all peer state to unknown
        resetNetworkInfo();

        // set own state to disconnected
        localInfo->State = peerInfo::disconnected;

        printf("successfully left the network\n");
    } else {
        return errNotConnected;
    }

    return SAFE;
    
}

int Peer::getSystemTime() {
    time_t seconds = time(0);
    DBG_PRINTF(("getSystemTime: %i\n", (int)seconds));

    return (int)seconds;
}


int Peer::create(string filename){
    int status = SAFE;
    int sysTime = getSystemTime();
    string shortName = VFS->getShortName(filename);
    if(!VFS->exist(shortName, 0)){
        ifstream ifile(filename.c_str());
        if (ifile) {
            status = VFS->insert(filename, sysTime);
            if (status == SAFE) {
                Job *job = new Job();
                NewModifiedArgs *jobArgs = new NewModifiedArgs();
                jobArgs->fileName = shortName;
                jobArgs->tagNumber = 0;
                jobArgs->timestamp = sysTime;       

                job->type = Job::NEWMODIFIED;
                job->args = (void *)jobArgs;

                pushJob(job);
            }

            printf("successfully created %s\n", shortName.c_str());
        }else{
            status = errNoFileFound;
        } 
    }else{
        status = errFileAlreadyExist;
    }
    return status;
}

int Peer::modify(string filename){
    return Peer::modify(filename, 0);
}

int Peer::modify(string filename, int version){
    int status = SAFE;

    if ( !VFS->exist(filename, version) ) {
        printf("%s does not exist.\n", filename.c_str());
        status = errNoFileFound;
    } else {
        // [**] make sure file is opened??

        status = VFS->modify(filename, version);
    }
    return status;
}

int Peer::closeFile(string filename){
    return Peer::closeFile(filename, 0);
}

int Peer::closeFile(string filename, int version){
    // [**] set status to closed??

    printf("%s is closed.\n", filename.c_str());

    return SAFE;
}

int Peer::openFile(string filename){
    return Peer::openFile(filename, 0);
}

int Peer::openFile(string filename, int version) {
    int status = SAFE;
    if (localInfo->State == peerInfo::connected) {
        // check if up to date
        if ( !VFS->exist(filename, version) ) {
            printf("%s does not exist.\n", filename.c_str());
            status = errNoFileFound;
        } else {
            if ( !VFS->isUpToDate(filename, version) ) {
                if ( !VFS->isVersionInTransit(filename, version) ) {
                    VFS->setVersionInTransit(filename, version);

                    // active pull
                    char msg[bufSize];
                    int msgLen;

                    RequestUpdateArgs *args = new RequestUpdateArgs();
                    args->fileName = filename;
                    args->tagNumber = version;

                    create_message(Peer::REQUESTUPDATE, args, msg, msgLen);

                    delete args;

                    // get file hosting information
                    Host *hostInfo = VFS->getHost(filename, version);

                    blockingSend(hostInfo->ip(), hostInfo->port(), msg, msgLen);
                } else {
                    // loop until finished passive pull
                    while ( VFS->isVersionInTransit(filename, version) ) {}
                }
            } 
        }
    } else {
        if ( !VFS->hasLocalCopy(filename, version) ) {
            printf("%s is unavailable.\n", filename.c_str());
            status = errVersionUnavailable;
        } 
    }

    if (status == SAFE) {
        printf("%s is open.\n", filename.c_str());
    }

    return status;
}

int Peer::deleteFile(string filename) {
    int status = SAFE;
    if ( !VFS->exist(filename, 0) ) {
        printf("%s does not exist.\n", filename.c_str());
        status = errNoFileFound;
    } else {
        int sysTime = getSystemTime();
        status = VFS->deleteFile(filename, sysTime);

        if (status == SAFE) {
            Job *job = new Job();
            DeleteFileArgs *jobArgs = new DeleteFileArgs();
            jobArgs->fileName = filename;
            jobArgs->timestamp = sysTime;       

            job->type = Job::DELETEFILE;
            job->args = (void *)jobArgs;

            pushJob(job);
        }
    }
    return status;
}

int Peer::save(string filename){
    return Peer::save(filename, 0);
}

int Peer::save(string filename, int version){
    int status = SAFE;

    if ( !VFS->exist(filename, version) ) {
        printf("%s does not exist.\n", filename.c_str());
        status = errNoFileFound;
    } else {
        if ( !VFS->isModified(filename, version) ) {
            printf("%s has not been modified.\n", filename.c_str());
            status = errFileNotModified;
        } else {

            int sysTime = getSystemTime();

            // [**] make sure file is opened??

            status = VFS->save(filename, version, sysTime);

            // only broadcast if on the network
            if (status == SAFE && localInfo->State == peerInfo::connected) {
                Job *job = new Job();
                NewModifiedArgs *jobArgs = new NewModifiedArgs();
                jobArgs->fileName = filename;
                jobArgs->tagNumber = version;
                jobArgs->timestamp = sysTime;       

                job->type = Job::NEWMODIFIED;
                job->args = (void *)jobArgs;

                pushJob(job);
            }
        }
    }
    return status;
}

int Peer::tag(string filename){
    return Peer::tag(filename, 0);
}


int Peer::tag(string filename, int version){
    int status = SAFE;

    if (localInfo->State != peerInfo::connected) {
        printf("cannot tag file when not on network\n");
        status = errNotConnected;
    } else {
        if ( !VFS->exist(filename, version) ) {
            printf("%s does not exist.\n", filename.c_str());
            status = errNoFileFound;
        } else {
            if ( VFS->isModified(filename, version) ) {
                printf("must save changes to file before tagging.\n");
                status = errFileNotModified;
            } else {
                int sysTime = getSystemTime();

                // [**] make sure file is opened??

                int newVersion = VFS->tag(filename, version, sysTime);

                if (status == SAFE) {
                    Job *job = new Job();
                    NewModifiedArgs *jobArgs = new NewModifiedArgs();
                    jobArgs->fileName = filename;
                    jobArgs->tagNumber = newVersion;
                    jobArgs->timestamp = sysTime;       

                    job->type = Job::NEWMODIFIED;
                    job->args = (void *)jobArgs;

                    pushJob(job);
                }
            }
        }
    }

    return status;
}

bool Peer::startListenerThread() {
    int status = SAFE;
    
    status = pthread_create(&_listenerContext, NULL, listener, this);

    return (status == SAFE);
}

bool Peer::startWorkerThreads() {
    int status = SAFE;
    
    for (int ii = 0; ii < maxWorkers; ii++) {
        pthread_t * newThreadContext = new pthread_t;
        status |= pthread_create(newThreadContext, NULL, worker, this);
        _workerContext.push_back(newThreadContext);
    }

    return (status == SAFE);
}

bool Peer::startUpdaterThread() {
    int status = SAFE;
    
    updater_terminate_flag = false;

    status = pthread_create(&_updaterContext, NULL, updater, this);

    return (status == SAFE);
}

bool Peer::terminateUpdaterThread() {
    int status = SAFE;
    
    updater_terminate_flag = true;

    return (status == SAFE);
}

void * Peer::listener(void *This) {
    ((Peer *)This)->listenerImplementation();
    return NULL;
     
}

void * Peer::worker(void *This) {
    ((Peer *)This)->workerImplementation();

    return NULL;
}

void * Peer::updater(void *This) {
    ((Peer *)This)->updaterImplementation();

    return NULL;
}


int Peer::updaterImplementation() {
    DBG_PRINTF(("starting updater\n"));

    while (!updater_terminate_flag) {
        // call VFS to get next file to pull
        string filename;
        int version;
        Host *hostInfo = VFS->getNextFileToPull(filename, version);
        if (hostInfo != NULL) {
            // passive pull
            char msg[bufSize];
            int msgLen;

            RequestUpdateArgs *args = new RequestUpdateArgs();
            args->fileName = filename;
            args->tagNumber = version;

            create_message(Peer::REQUESTUPDATE, args, msg, msgLen);

            delete args;

            blockingSend(hostInfo->ip(), hostInfo->port(), msg, msgLen);
        }
    }   

    DBG_PRINTF(("terminating updater\n"));

    return 0;
}


int Peer::workerImplementation() {
    Job *job = NULL;

    while (true) {
        sem_wait (&job_queue_count);
        pthread_mutex_lock (&job_queue_mutex);
        if( !jobs.empty() ){
            job = jobs.front();
            DBG_PRINTF(("retrieving job...\n"));
            jobs.pop(); 
        }
        pthread_mutex_unlock (&job_queue_mutex);

        int status = process_job(job);

        if(status == exitWorker)break;
    }
    DBG_PRINTF(("WORKER EXITS\n"));

    return 0;
}

int Peer::listenerImplementation() {
    DBG_PRINTF(("starting listener\n"));

    int status = SAFE;

    // opens server socket
    int         server_sock,          // Original socket in server
                    client_sock;           // New socket from connect
    socklen_t       clnt_len;           // Length of client address
    struct sockaddr_in                  // Internet addr client & server
                    clnt_adr, serv_adr;
    int             len;             // Misc counters, etc.

    // SOCKET creation 
    if ((server_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        perror("generate error");
    }

    memset( &serv_adr, 0, sizeof(serv_adr) );      // Clear structure
    serv_adr.sin_family      = AF_INET;            // Set address type
    serv_adr.sin_addr.s_addr = htonl(INADDR_ANY);  // Any interface
    serv_adr.sin_port        = htons( localInfo->Port );        // Use our fake port

    // BIND
    while (bind( server_sock, (struct sockaddr *) &serv_adr,
                sizeof(serv_adr)) < 0){
        perror("bind error");
        DBG_PRINTF(("listener: trying to bind again...\n"));
    }

    // LISTEN
    if (listen(server_sock, 5) < 0 ) {
        perror("listen error");
        close (server_sock);
        return 4;
    }                   
    
        // accepts incoming client msgs
    do {
        DBG_PRINTF(("listening indefinitely\n"));

        clnt_len = sizeof(clnt_adr); 
        if ((client_sock = accept( server_sock, (struct sockaddr *) &clnt_adr,
                        &clnt_len)) < 0) {
            perror("accept error");
            close(server_sock);
            return 5;
        }

        // read from client socket
        char buf[bufSize];
        if ( (len=read(client_sock, buf, bufSize)) > 0 ){
            DBG_PRINTF(("received msg %s\n", buf));
            status = process_message (client_sock, buf, len);
            if ( status == exitListener ) {
                break;
            } else if ( status > 0 ) {
                printf("process_message() returned WARNING 0x%x\n", status);
            }
        }
    } while( 1 );

    close(server_sock);

    DBG_PRINTF(("terminating listener\n"));

    return status;
}


int Peer::create_message(Peer::Message type, void *msgParams, char *mail, int &length){

    length = 0;

    if(mail != NULL) {
        switch(type){
            case JOINANDQUERY:
                length += sprintf(mail, "^%i%%%s%%%i%%%s$", type, localInfo->IP.c_str(), localInfo->Port, localInfo->deviceType.c_str());
                break;
            case QUERYVFS:    
            case NAK: 
            case LEAVE:        
                length += sprintf(mail, "^%i%%%s%%%i$", type, localInfo->IP.c_str(), localInfo->Port);
                break;
            case REPLYSTATUS:
                length += sprintf(mail, "^%i%%%s%%%i%%%s$", type, localInfo->IP.c_str(), localInfo->Port, localInfo->deviceType.c_str());
                break;
            case REPLYVFS: {
                ReplyVFSArgs *params = (ReplyVFSArgs *)msgParams;
                if (params != NULL) {     
                    length += sprintf(mail, "^%i%%%s%%%i%%%s$", type, localInfo->IP.c_str(), localInfo->Port, params->message.c_str());
                }
                break;
            }
            case QUERYTRANSFERFILES: {
                string msgBody = "";      
                VFS->getHostedFiles(msgBody);
                length += sprintf(mail, "^%i%%%s%%%i%%%s$", type, localInfo->IP.c_str(), localInfo->Port, msgBody.c_str());
                break;
            }
            case NEWMODIFIED: {
                NewModifiedArgs *params = (NewModifiedArgs *)msgParams;
                if (params != NULL) {     
                    length += sprintf(mail, "^%i%%%s%%%i%%%s%%%i%%%i$", type, localInfo->IP.c_str(), localInfo->Port, 
                                                                        params->fileName.c_str(), params->tagNumber, params->timestamp);
                }
                break;
            }
            case DELETEDFILE: {
                DeleteFileArgs *params = (DeleteFileArgs *)msgParams;
                if (params != NULL) {     
                    length += sprintf(mail, "^%i%%%s%%%i%%%s%%%i$", type, localInfo->IP.c_str(), localInfo->Port, 
                                                                        params->fileName.c_str(), params->timestamp);
                }
                break;
            }
            case REPLYTRANSFERFILES: {
                ReplyTransferFilesArgs *params = (ReplyTransferFilesArgs *)msgParams;
                if (params != NULL) {
                    vector<string>required_files;
                    vector<int>required_versions;

                    for (int i = 0; i < (int)params->fileNames.size(); i++) {
                        string fileName = params->fileNames[i];
                        int tagNumber = params->versions[i];

                        if ( !VFS->isUpToDate(fileName, tagNumber) && !VFS->isVersionInTransit(fileName, tagNumber) ) {
                            // set inTransit bit since we are expecting the file                            
                            VFS->setVersionInTransit(fileName, tagNumber);

                            required_files.push_back(fileName);
                            required_versions.push_back(tagNumber);
                        }
                    }

                    length += sprintf(mail, "^%i%%%s%%%i%%%i", type, localInfo->IP.c_str(), localInfo->Port, (int)required_files.size());

                    for (int i = 0; i < (int)required_files.size(); i++) {
                        length += sprintf(mail + length, "%%%s%%%i", required_files[i].c_str(), required_versions[i]);
                    }
                                        
                    length += sprintf(mail + length, "$");

                    // update args to include only the required files
                    params->fileNames = required_files;
                    params->versions = required_versions;
                }
                break;
            }
            case TRANSFERFILE: {
                TransferFileArgs *params = (TransferFileArgs *)msgParams;
                if (params != NULL) {
                    length += sprintf(mail, "^%i%%%s%%%i%%%s%%%i%%%i%%%i$", type, localInfo->IP.c_str(), localInfo->Port, 
                                                                            params->fileName.c_str(), params->tagNumber, params->timestamp, params->chunks);
                }
                break;
            }
            case REQUESTUPDATE: {
                RequestUpdateArgs *params = (RequestUpdateArgs *)msgParams;
                if (params != NULL) {
                    length += sprintf(mail, "^%i%%%s%%%i%%%s%%%i$", type, localInfo->IP.c_str(), localInfo->Port, params->fileName.c_str(), params->tagNumber);
                }
                break;
            }
            case EXITLISTENER:
                length = sprintf(mail, "^%i%%%s%%%i$", type, localInfo->IP.c_str(), localInfo->Port);
                break;
            default:
                cout << "invalid jobtype: " << type << endl;
        }
    }

    // add null terminator string length
    length++;

    DBG_PRINTF(("created message: %s with length %i\n", mail, length));

    return SAFE;
}

int Peer::process_message(int client_sock, char *buf, int bufLen) {
    int status = SAFE;

    // parse buf
    char *field, *brkt;
    int msgType;
    string ip = "";
    int port = 0;

    if (bufLen < 3 || buf[0] != '^' || buf[bufLen-2] != '$') {
        DBG_PRINTF(("message: %s with length = %i\n", buf, bufLen));   
        return errInvalidMessage; // return errInvalidMessage error
    }

    field = strtok_r(buf, "^%$", &brkt); // get message type
    msgType = atoi(field);
    DBG_PRINTF(("msg type: %i\n", msgType));

    // continue parsing
    field = strtok_r(NULL, "^%$", &brkt); // get IP addr
    ip += field;

    field = strtok_r(NULL, "^%$", &brkt); // get Port #
    port = atoi(field);

    switch(msgType){
        /*** server receives these messages ***/
        case REQUESTUPDATE: {
            string fileName = "";
            int version;

            field = strtok_r(NULL, "^%$", &brkt); // get file name
            fileName += field;
    
            field = strtok_r(NULL, "^%$", &brkt); // get version #
            version = atoi(field);

            vector<string>phys_chunks;
            int timestamp;

            VFS->getPhysicalFile(fileName, version, phys_chunks, timestamp);

            Job *job = new Job();
            TransferFileArgs *jobArgs = new TransferFileArgs();        
            jobArgs->clntSocket = client_sock;
            jobArgs->fileName = fileName;
            jobArgs->tagNumber = version;
            jobArgs->timestamp = timestamp;
            jobArgs->chunks = phys_chunks.size();

            job->type = Job::TRANSFERFILE;
            job->args = (void *)jobArgs;

            pushJob(job);

            break;
        }
        case JOINANDQUERY: {
            string deviceType = "";

            field = strtok_r(NULL, "^%$", &brkt); // get device Type
            deviceType += field;

            DBG_PRINTF(("received JOIN: %s:%i with deviceType = %s\n", ip.c_str(), port, deviceType.c_str()));

            Job *job = new Job();
            if (localInfo->State == peerInfo::connected) {

                // create REPLYSTATUS job
                QueryStatusArgs *jobArgs = new QueryStatusArgs();        
                jobArgs->clntSocket = client_sock;

                job->type = Job::REPLYSTATUS;
                job->args = (void *)jobArgs;

                // update network status  
                peerJoin(ip, port, deviceType);
            } else {
                // create NAK job
                NakArgs *jobArgs = new NakArgs();        
                jobArgs->clntSocket = client_sock;

                job->type = Job::NAK;
                job->args = (void *)jobArgs;
            }

            // acquire the job queue mutex; enqueue; release mutex
            pushJob(job);

            break;
        }
        case EXITLISTENER:
            close(client_sock);
            return exitListener;
        case QUERYVFS: {
            string tempMsgBuffer = "";

            VFS->getFileSystem(localInfo->IP, localInfo->Port, tempMsgBuffer); 

            Job *job = new Job();
            ReplyVFSArgs *jobArgs = new ReplyVFSArgs();
            jobArgs->clntSocket = client_sock;
            jobArgs->message = tempMsgBuffer;
            
            job->type = Job::REPLYVFS;
            job->args = (void *)jobArgs;

            pushJob(job);

            break;
        }
        case LEAVE: {
            close(client_sock);

            VFS->removeHost(ip, port);

            // update network status  
            peerLeave(ip, port);

            break;
        }

        /*** client receives these messages ***/
        case REPLYSTATUS: {
            close(client_sock);

            string deviceType = "";

            field = strtok_r(NULL, "^%$", &brkt); // get device Type
            deviceType += field;

            DBG_PRINTF(("received REPLYSTATUS: %s:%i with deviceType = %s\n", ip.c_str(), port, deviceType.c_str()));

            peerJoin(ip, port, deviceType);

            break;
        }
        case QUERYTRANSFERFILES: {
            int num_files_queried = 0;

            field = strtok_r(NULL, "^%$", &brkt); // get num_files_queried
            num_files_queried = atoi(field);

            DBG_PRINTF(("received QUERYTRANSFERFILES: %s:%i\n", ip.c_str(), port));

            Job *job = new Job();
            ReplyTransferFilesArgs *jobArgs = new ReplyTransferFilesArgs();

            for (int i = 0; i < num_files_queried; i++) {
                string fileName = "";
                int version;

                field = strtok_r(NULL, "^%$", &brkt); // get file name
                fileName += field;
        
                field = strtok_r(NULL, "^%$", &brkt); // get version #
                version = atoi(field);

                jobArgs->fileNames.push_back(fileName);
                jobArgs->versions.push_back(version);
            }
            
            jobArgs->clntSocket = client_sock;
            job->type = Job::REPLYTRANSFERFILES;
            job->args = (void *)jobArgs;

            pushJob(job);

            break;
        }
        case REPLYTRANSFERFILES: {
            int num_files_needed = 0;

            field = strtok_r(NULL, "^%$", &brkt); // get num_files_needed
            num_files_needed = atoi(field);

            DBG_PRINTF(("received REPLYTRANSFERFILES: %s:%i\n", ip.c_str(), port));

            for (int i = 0; i < num_files_needed; i++) {
                string fileName = "";
                int version, timestamp;

                field = strtok_r(NULL, "^%$", &brkt); // get file name
                fileName += field;
        
                field = strtok_r(NULL, "^%$", &brkt); // get version #
                version = atoi(field);

                vector<string>phys_chunks;
                VFS->getPhysicalFile(fileName, version, phys_chunks, timestamp);

                int msgLen = 0;
                char msg[bufSize];

                TransferFileArgs *jobArgs = new TransferFileArgs();
                jobArgs->clntSocket = client_sock;
                jobArgs->fileName = fileName;
                jobArgs->tagNumber = version;
                jobArgs->timestamp = timestamp;
                jobArgs->chunks = phys_chunks.size();

                create_message(Peer::TRANSFERFILE, (void *)jobArgs, msg, msgLen);

                delete jobArgs;

                DBG_PRINTF(("about to write %s with length %i\n", msg, msgLen));
                write(client_sock, msg, msgLen);

                // get a ACK back before sending the chunks over
                int responseLen = 0;
                char buf[3];
                if (( responseLen = read(client_sock, buf, 3)) > 0 ) {
                    if ( buf[0] != '^' || buf[1] != '$' ) {
                        DBG_PRINTF(("chunk transfer not ACKed!!!\n"));
                        return errUnknownWarning;
                    }
                }

                for (int j = 0; j < (int)phys_chunks.size(); j++) {
                    write(client_sock, phys_chunks[j].c_str(), phys_chunks[j].length());

                    if (( responseLen = read(client_sock, buf, 3)) > 0 ) {
                        if ( buf[0] != '^' || buf[1] != '$' ) {
                            DBG_PRINTF(("chunk transfer not ACKed!!!\n"));
                            return errUnknownWarning;
                        } else {
                            DBG_PRINTF(("chunk %i transfer ACKed: %s!!!\n", j, buf));
                        }
                    }
                }

            }

            close(client_sock);

            break;
        }
        case NAK: {
            close(client_sock);

            DBG_PRINTF(("received NAK: %s:%i\n", ip.c_str(), port));

            // peer doesn't want to be bothered, set its state to be disconnected
            peerLeave(ip, port);

            break;
        }
        case REPLYVFS: {
            close(client_sock);

            DBG_PRINTF(("received REPLYVFS: %s:%i\n", ip.c_str(), port));

            field = strtok_r(NULL, "$", &brkt); // get contents

            vector<FileInfo *>deletedFiles = VFS->SyncFileSystem(field);

            for (int i = 0; i < (int)deletedFiles.size(); i++) {
                Job *job = new Job();

                DeleteFileArgs *jobArgs = new DeleteFileArgs();        
                jobArgs->fileName = deletedFiles[i]->fileName();
                jobArgs->timestamp = deletedFiles[i]->timestamp();

                job->type = Job::DELETEFILE;
                job->args = (void *)jobArgs;

                pushJob(job);
            }

            // free FileInfo* vector
            for (int i = 0; i < (int)deletedFiles.size(); i++) {
                FileInfo *temp = deletedFiles.back();            
                deletedFiles.pop_back();
                delete temp;
            }

            vector<FileInfo *>modifiedFiles = VFS->filesToBroadcast();
            DBG_PRINTF(("filesToBroadcast returned %i many files", (int)modifiedFiles.size()));

            for (int i = 0; i < (int)modifiedFiles.size(); i++) {
                Job *job = new Job();

                NewModifiedArgs *jobArgs = new NewModifiedArgs();        
                jobArgs->fileName = modifiedFiles[i]->fileName();
                jobArgs->tagNumber = modifiedFiles[i]->version();
                jobArgs->timestamp = modifiedFiles[i]->timestamp();

                job->type = Job::NEWMODIFIED;
                job->args = (void *)jobArgs;

                pushJob(job);
            }

            // free FileInfo* vector
            for (int i = 0; i < (int)modifiedFiles.size(); i++) {
                FileInfo *temp = modifiedFiles.back();            
                modifiedFiles.pop_back();
                delete temp;
            }

            break;
        }
        case TRANSFERFILE: {
            // DO NOT CLOSE SOCKET, may be more files coming
            string fileName;
            int tagNumber, timestamp;
            int chunks;

            field = strtok_r(NULL, "^%$", &brkt); // file name
            fileName += field;

            field = strtok_r(NULL, "^%$", &brkt); // get version
            tagNumber = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get time stamp
            timestamp = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get number of chunks
            chunks = atoi(field);

            int responseLen = 0;
            char buf[bufSize];

            // send ACK to tell sender to start sending chunks
            write(client_sock, "^$", 3);

            for (int i = 1; i <= chunks; i++) {
                DBG_PRINTF(("expecting chunks: %i\n", i));
                if (( responseLen = read(client_sock, buf, bufSize)) > 0 ) {
                    DBG_PRINTF(("chunk length: %i with bufSize %i\n", responseLen, bufSize));
                    VFS->writeToDisk(fileName, tagNumber, i, buf, responseLen);
                }

                // send ACK
                write(client_sock, "^$", 3);
            }
            VFS->combineFileOnDisk(fileName, tagNumber, chunks);

            VFS->receivedFile(fileName, tagNumber, timestamp);
            break;
        }
        case NEWMODIFIED: {
            close(client_sock);

            string fileName = "";
            int tagNumber, timestamp;

            field = strtok_r(NULL, "^%$", &brkt); // file name
            fileName += field;

            field = strtok_r(NULL, "^%$", &brkt); // get tagnumber
            tagNumber = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get timestamp
            timestamp = atoi(field);

            Host *host = new Host(ip, port);
            DBG_PRINTF(("host created with ip %s port %i\n", ip.c_str(), port));
            VFS->updateNetworkTimestamp(fileName, tagNumber, host, timestamp);

            break;
        }
        case DELETEDFILE: {
            close(client_sock);

            string fileName = "";
            int timestamp;

            field = strtok_r(NULL, "^%$", &brkt); // file name
            fileName += field;

            field = strtok_r(NULL, "^%$", &brkt); // get timestamp
            timestamp = atoi(field);

            VFS->deleteFile(fileName, timestamp);

            break;
        }
        default:
            close(client_sock);
            cout<<"invalid jobtype: "<< msgType <<endl;
            break;
    }
    
    DBG_PRINTF(("safely processed message\n"));
    return status;

}

int Peer::blockingSend(string ip, int port, char *mail, int length) {
    // [*] assert ip and port are on connected peerlist

    // uses socket connection to send to all peers in network
    DBG_PRINTF(("Peer::blockingSend: %s\n", mail));

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info

    host = gethostbyname( ip.c_str() );       // Obtain host (server) info
    if (host == (struct hostent *) NULL ) {
        DBG_PRINTF(("host name not found for %s:%i\n", ip.c_str(), port));
        perror("gethostbyname ");
    } else {
        memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
        serv_adr.sin_family = AF_INET;                 // Set address type
        memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
        serv_adr.sin_port   = htons( port );

        if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
            perror("generate error");
            return 3;
        } 

        // try to connect to another peer (this is blocking???)
        if (connect( orig_sock,(struct sockaddr *)&serv_adr, sizeof(serv_adr)) < 0) {
            DBG_PRINTF(("cannot connect to peer %s:%i\n", ip.c_str(), port));
            //perror("connect error");
        } else {
            // write message
            write(orig_sock, mail, length);

            // wait for reply
            int responseLen = 0;
            char buf[bufSize];
            if (( responseLen = read(orig_sock, buf, bufSize)) > 0 ) {
                // process_message either closes socket or stores it in a job arg and closes it once job is completed
                process_message(orig_sock, buf, responseLen);
            }
        }
    }

    return SAFE;
}

int Peer::nonblockingSend(string ip, int port, char *mail, int length) {
    // [*] assert ip and port are on connected peerlist

    // uses socket connection to send to all peers in network
    DBG_PRINTF(("Peer::nonblockingSend: %s\n", mail));

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info

    host = gethostbyname( ip.c_str() );       // Obtain host (server) info
    if (host == (struct hostent *) NULL ) {
        DBG_PRINTF(("host name not found for %s:%i\n", ip.c_str(), port));
        perror("gethostbyname ");
    } else {
        memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
        serv_adr.sin_family = AF_INET;                 // Set address type
        memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
        serv_adr.sin_port   = htons( port );

        if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
            perror("generate error");
            return 3;
        } 

        // try to connect to another peer (this is blocking???)
        if (connect( orig_sock,(struct sockaddr *)&serv_adr, sizeof(serv_adr)) < 0) {
            DBG_PRINTF(("cannot connect to peer %s:%i\n", ip.c_str(), port));
            //perror("connect error");
        } else {
            // write message
            write(orig_sock, mail, length);

            close(orig_sock);
        }
    }

    return SAFE;
}

int Peer::processPeerList(string peersFile) {
    int peer_num = 0;

    ifstream myfile(peersFile.c_str());
    if (myfile.is_open()) {

        // parse peersFile
        string ip_addr;
        int port_num;
        while ( myfile >> ip_addr >> port_num )
        {
            DBG_PRINTF(("peer %i is %s:%i\n", peer_num, ip_addr.c_str(), port_num));
            _peers[peer_num] = new peerInfo();
            _peers[peer_num]->IP = ip_addr;
            _peers[peer_num]->Port = port_num;
            _peers[peer_num]->State = peerInfo::unknown;
            _peers[peer_num]->deviceType = "unknown";

            peer_num++;

            // [*] error check to make sure num_peers does not exceed max
        }
        myfile.close();
    }
    else return errNoFileFound;

    _numPeers = peer_num;

    return SAFE;
}

int Peer::blockingBroadcast(char *mail, int length) {
    // uses socket connection to send to all peers in network
    DBG_PRINTF(("Peer::blockingBroadcast: %s\n", mail));

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info

    // loop through all online peers and tell them I have arrived!
    // create thread per peer
    for (int ii = 0; ii < _numPeers; ii++) {

        pthread_mutex_lock (&network_info_mutex);
        peerInfo::Connection curState = _peers[ii]->State;
        string IP = _peers[ii]->IP;
        int port = _peers[ii]->Port;
        pthread_mutex_unlock (&network_info_mutex);

        if ( curState != peerInfo::disconnected ) {
            host = gethostbyname( IP.c_str() );       // Obtain host (server) info
            if (host == (struct hostent *) NULL ) {
                DBG_PRINTF(("host name not found for %s:%i\n", IP.c_str(), port));
                perror("gethostbyname ");
            } else {
                memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
                serv_adr.sin_family = AF_INET;                 // Set address type
                memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
                serv_adr.sin_port   = htons( port );

                if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                    perror("generate error");
                    return 3;
                } 

                // try to connect to another peer (this is blocking???)
                if (connect( orig_sock,(struct sockaddr *)&serv_adr, sizeof(serv_adr)) < 0) {
                    DBG_PRINTF(("cannot connect to peer %s:%i\n", IP.c_str(), port));
                    //perror("connect error");
                } else {
                    // write message
                    write(orig_sock, mail, length);

                    // wait for reply
                    int responseLen = 0;
                    char buf[bufSize];
                    if (( responseLen = read(orig_sock, buf, bufSize)) > 0 ) {
                        // process_message either closes socket or stores it in a job arg and closes it once job is completed
                        process_message(orig_sock, buf, responseLen);
                    }

                }
            }
        }
    }

    DBG_PRINTF(("BROADCAST DONE\n"));

    return SAFE;
}

int Peer::nonblockingBroadcast(char *mail, int length) {
    // uses socket connection to send to all peers in network
    DBG_PRINTF(("Peer::nonblockingBroadcast: %s\n", mail));

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info

    // loop through all online peers and tell them I have arrived!
    // create thread per peer
    for (int ii = 0; ii < _numPeers; ii++) {

        pthread_mutex_lock (&network_info_mutex);
        peerInfo::Connection curState = _peers[ii]->State;
        string IP = _peers[ii]->IP;
        int port = _peers[ii]->Port;
        pthread_mutex_unlock (&network_info_mutex);

        if ( curState == peerInfo::connected ) {
            host = gethostbyname( IP.c_str() );       // Obtain host (server) info
            if (host == (struct hostent *) NULL ) {
                DBG_PRINTF(("host name not found for %s:%i\n", IP.c_str(), port));
                perror("gethostbyname ");
            } else {
                memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
                serv_adr.sin_family = AF_INET;                 // Set address type
                memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
                serv_adr.sin_port   = htons( port );

                if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                    perror("generate error");
                    return 3;
                } 

                // try to connect to another peer (this is blocking???)
                if (connect( orig_sock,(struct sockaddr *)&serv_adr, sizeof(serv_adr)) < 0) {
                    DBG_PRINTF(("cannot connect to peer %s:%i\n", IP.c_str(), port));
                    //perror("connect error");
                } else {
                    // write message
                    write(orig_sock, mail, length);

                    // does not need to wait for reply...
                    close(orig_sock);

                }
            }
        }
    }

    DBG_PRINTF(("BROADCAST DONE\n"));

    return SAFE;
}

int Peer::blockingSendToAny(char *mail, int length) {
    // uses socket connection to send to all peers in network
    DBG_PRINTF(("Peer::blockingSendToAny: %s\n", mail));

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info

    // loop through all online peers and tell them I have arrived!
    // create thread per peer
    for (int ii = 0; ii < _numPeers; ii++) {

        pthread_mutex_lock (&network_info_mutex);
        peerInfo::Connection curState = _peers[ii]->State;
        string IP = _peers[ii]->IP;
        int port = _peers[ii]->Port;
        pthread_mutex_unlock (&network_info_mutex);

        if ( curState == peerInfo::connected ) {
            host = gethostbyname( IP.c_str() );       // Obtain host (server) info
            if (host == (struct hostent *) NULL ) {
                DBG_PRINTF(("host name not found for %s:%i\n", IP.c_str(), port));
                perror("gethostbyname ");
            } else {
                memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
                serv_adr.sin_family = AF_INET;                 // Set address type
                memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
                serv_adr.sin_port   = htons( port );

                if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                    perror("generate error");
                    return 3;
                } 

                // try to connect to another peer (this is blocking???)
                if (connect( orig_sock,(struct sockaddr *)&serv_adr, sizeof(serv_adr)) < 0) {
                    DBG_PRINTF(("cannot connect to peer %s:%i\n", IP.c_str(), port));
                    //perror("connect error");
                } else {
                    // write message
                    write(orig_sock, mail, length);

                    // wait for reply
                    int responseLen = 0;
                    char buf[bufSize];
                    if (( responseLen = read(orig_sock, buf, bufSize)) > 0 ) {
                        // process_message either closes socket or stores it in a job arg and closes it once job is completed
                        process_message(orig_sock, buf, responseLen);
                    }

                    // exit when received a response
                    break;

                }
            }
        }
    }

    DBG_PRINTF(("BROADCAST DONE\n"));

    return SAFE;
}

int Peer::pushJob(Job * job){
    pthread_mutex_lock (&job_queue_mutex);
    DBG_PRINTF(("pushing job into queue: %i\n", job->type));
    jobs.push(job);
    sem_post (&job_queue_count);
    pthread_mutex_unlock (&job_queue_mutex);
    return SAFE;
}

int Peer::process_job(Job *job) {
    int status = SAFE;

    // depending on job type, connect to peer mailbox or just reply using passed in client socket

    // parse job args depending on job type
    switch(job->type){
        /*** SERVER JOBS ***/
        case Job::REPLYVFS: {

            int msgLen = 0;            
            char msg[bufSize];

            DBG_PRINTF(("processing REPLYVFS job...\n"));

            create_message(Peer::REPLYVFS, job->args, msg, msgLen);

            write( ((ReplyVFSArgs *)(job->args))->clntSocket, msg, msgLen );

            break;
        }
        case Job::REPLYSTATUS: {

            int msgLen = 0;            
            char msg[bufSize];

            DBG_PRINTF(("processing REPLYSTATUS job...\n"));

            create_message(Peer::REPLYSTATUS, job->args, msg, msgLen);

            write( ((QueryStatusArgs *)(job->args))->clntSocket, msg, msgLen );

            break;
        }
        case Job::TRANSFERFILE: {

            int msgLen = 0; 
            int timestamp = 0;           
            char msg[bufSize];

            DBG_PRINTF(("processing TRANSFERFILE job...\n"));

            create_message(Peer::TRANSFERFILE, job->args, msg, msgLen);

            write( ((TransferFileArgs *)(job->args))->clntSocket, msg, msgLen );

            char buf[3];
            int responseLen = 0;
            if (( responseLen = read(((TransferFileArgs *)(job->args))->clntSocket, buf, 3)) > 0 ) {
                if ( buf[0] != '^' || buf[1] != '$' ) {
                    DBG_PRINTF(("chunk transfer not ACKed!!!\n"));
                    return errUnknownWarning;
                }
            } 

            vector<string>phys_chunks;
            VFS->getPhysicalFile(((TransferFileArgs *)(job->args))->fileName, 
                            ((TransferFileArgs *)(job->args))->tagNumber, phys_chunks, timestamp);


            for (int j = 0; j < (int)phys_chunks.size(); j++) {
                DBG_PRINTF(("writing chunk %i chunk length: %i \n", j+1, (int)phys_chunks[j].length()));
                write(((TransferFileArgs *)(job->args))->clntSocket, phys_chunks[j].c_str(), (int)phys_chunks[j].length());

                if (( responseLen = read(((TransferFileArgs *)(job->args))->clntSocket, buf, 3)) > 0 ) {
                    if ( buf[0] != '^' || buf[1] != '$' ) {
                        DBG_PRINTF(("chunk transfer not ACKed!!!\n"));
                        return errUnknownWarning;
                    }
                } 
            }


            break;
        }
        case Job::NAK: {

            int msgLen = 0;            
            char msg[bufSize];

            DBG_PRINTF(("processing NAK job...\n"));

            create_message(Peer::NAK, job->args, msg, msgLen);

            write( ((NakArgs *)(job->args))->clntSocket, msg, msgLen );

            break;
        }
        case Job::REPLYTRANSFERFILES: {
            // THIS MAKES SURE VFS is set up
            pthread_mutex_lock (&VFS_mutex);
            pthread_mutex_unlock (&VFS_mutex);

            int msgLen = 0;            
            char msg[bufSize];

            DBG_PRINTF(("processing REPLYTRANSFERFILES job...\n"));

            create_message(Peer::REPLYTRANSFERFILES, job->args, msg, msgLen);

            ReplyTransferFilesArgs *args = (ReplyTransferFilesArgs *)job->args;

            write( args->clntSocket, msg, msgLen );

            int responseLen = 0;
            char buf[bufSize];

            DBG_PRINTF(("need to transfer over %i files\n", (int)args->fileNames.size()));
            for (int i = 0; i < (int)args->fileNames.size(); i++) {
                if (( responseLen = read(args->clntSocket, buf, bufSize)) > 0 ) {
                    DBG_PRINTF(("received message %s with length %i\n", buf, responseLen));
                    process_message(args->clntSocket, buf, responseLen);
                }    
            }

            vector<FileInfo *>modifiedFiles = VFS->filesToBroadcast();

            for (int i = 0; i < (int)modifiedFiles.size(); i++) {
                Job *job = new Job();

                NewModifiedArgs *jobArgs = new NewModifiedArgs();        
                jobArgs->fileName = modifiedFiles[i]->fileName();
                jobArgs->tagNumber = modifiedFiles[i]->version();
                jobArgs->timestamp = modifiedFiles[i]->timestamp();

                job->type = Job::NEWMODIFIED;
                job->args = (void *)jobArgs;

                pushJob(job);
            }

            break;
        }

        /*** CLIENT JOBS ***/
        case Job::EXITLISTENER: {
            char msg[bufSize];
            int msgLen = 0;

            DBG_PRINTF(("processing EXITLISTENER job...\n"));

            // create NEWFILE message
            status = create_message(Peer::EXITLISTENER, job->args, msg, msgLen);

            // send to self function
            nonblockingSend(localInfo->IP, localInfo->Port, msg, msgLen);

            break;
        }
        case Job::NEWMODIFIED: {
            // THIS MAKES SURE VFS is set up
            pthread_mutex_lock (&VFS_mutex);
            pthread_mutex_unlock (&VFS_mutex);

            char msg[bufSize];
            int msgLen = 0;

            DBG_PRINTF(("processing NEWMODIFIED job...\n"));

            // create NEWMODIFIED message
            status = create_message(Peer::NEWMODIFIED, job->args, msg, msgLen);

            nonblockingBroadcast(msg, msgLen);

            break;
        }
        case Job::DELETEFILE: {
            // THIS MAKES SURE VFS is set up
            pthread_mutex_lock (&VFS_mutex);
            pthread_mutex_unlock (&VFS_mutex);

            char msg[bufSize];
            int msgLen = 0;

            DBG_PRINTF(("processing DELETEFILE job...\n"));

            // create DELETEDFILE message
            status = create_message(Peer::DELETEDFILE, job->args, msg, msgLen);

            nonblockingBroadcast(msg, msgLen);

            break;
        }
        case Job::EXITWORKER: {
            status = exitWorker;
            break;
        }
        default:
            status = errInvalidJob;  // error: unspecified job type
            break;
    }

    delete job;

    return status;
}

int Peer::peerLeave(string ip, int port) {
    pthread_mutex_lock (&network_info_mutex);

    for (int ii = 0; ii < _numPeers; ii++) {
        if (_peers[ii]->IP == ip && _peers[ii]->Port == port) {
            _peers[ii]->State = peerInfo::disconnected;
            break;
        }
    }

    pthread_mutex_unlock (&network_info_mutex);

    return SAFE;
}

int Peer::peerJoin(string ip, int port, string deviceType) {
    pthread_mutex_lock (&network_info_mutex);

    for (int ii = 0; ii < _numPeers; ii++) {
        if (_peers[ii]->IP == ip && _peers[ii]->Port == port) {
                _peers[ii]->State = peerInfo::connected;
                _peers[ii]->deviceType = deviceType;
                break;
        }
    } 

    pthread_mutex_unlock (&network_info_mutex);

    return SAFE;
}

int Peer::resetNetworkInfo() {
    pthread_mutex_lock (&network_info_mutex);

    for (int ii = 0; ii < _numPeers; ii++) {
        _peers[ii]->State = peerInfo::unknown;
    } 

    pthread_mutex_unlock (&network_info_mutex);

    return SAFE;
}

string convertInt(int number)
{
   stringstream ss;
   ss << number;
   return ss.str();
}

int Peer::test(){
    vector<string> chunks;
    ifstream ifile("p1.pdf");
        if(ifile){
            // get length of file:
            ifile.seekg (0, ios::end);
            long length = ifile.tellg();
            ifile.seekg (0, ios::beg);
                
            int chunkNum = length/chunkSize+1;

            chunks.clear();
            for(int i = 1; i <= chunkNum; i++){
                string output = "p1.part"+convertInt(i);
                ofstream ofile (output.c_str(), fstream::trunc|ofstream::binary);
                // read data as a block:
                DBG_PRINTF(("reading chunk %i length: %li chunkSize: %i \n", i, length, chunkSize));
                char data[chunkSize];
                if(length < chunkSize){
                    ifile.read(data, length);
                    string chunk (data, length);
                DBG_PRINTF(("strlen(data) returns %i \n", (int)strlen(data)));
                DBG_PRINTF(("reading chunk %i chunk length: %i \n", i, (int)chunk.length()));
                ofile.write(chunk.c_str(), length);
                }else{
                    ifile.read(data, chunkSize);
                    string chunk (data, chunkSize);
                DBG_PRINTF(("strlen(data) returns %i \n", (int)strlen(data)));
                DBG_PRINTF(("reading chunk %i chunk length: %i \n", i, (int)chunk.length()));
                ofile.write(chunk.c_str(), chunkSize);
                }
                
                ofile.close();
           
                length -= chunkSize;
            }

            string finaloutput = "p1copy.pdf";
             ofstream outfile (finaloutput.c_str(), fstream::trunc|ofstream::binary);

            for(int i = 1; i <= chunkNum; i++){
                string input = "p1.part"+convertInt(i);
                ifstream infile(input.c_str());
                if(infile){
                    outfile<<infile.rdbuf();/*
                    // remove the partial file
                    if( remove( input.c_str() ) != 0 ){
                        DBG_PRINTF(( "Error deleting file: %s\n", input.c_str() ));
                    } else {
                        DBG_PRINTF(( "File successfully deleted %s\n", input.c_str()  ));
                    }*/
                }else{
                    break;
                }
                infile.close();
            }
            outfile.close();
     
        }else{
            DBG_PRINTF(("cannot open physical file even though logical file exists\n"));
        }
        ifile.close();
    return SAFE;
}



